package longshu.srpc.server.impl;

import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import longshu.srpc.IOUtils;
import longshu.srpc.api.RpcProtocol;
import longshu.srpc.api.RpcRequest;
import longshu.srpc.api.RpcResponse;
import longshu.srpc.server.RpcAcceptor;
import longshu.srpc.server.RpcProcessor;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
public class SimpleRpcAcceptor implements RpcAcceptor {
    private static ExecutorService executor;
    private ServerSocket serverSocket;
    private String protocol;
    private String host;
    private int port;
    private String serverAddress;
    private volatile boolean running = false;

    @Setter
    private RpcProcessor rpcProcessor;

    public SimpleRpcAcceptor(String serverAddress) {
        this.serverAddress = serverAddress;

        URL url = null;
        try {
            url = new URL(serverAddress);
        } catch (MalformedURLException e) {
            throw new RuntimeException(e);
        }
        protocol = url.getProtocol();
        host = url.getHost();
        port = url.getPort();
    }

    public SimpleRpcAcceptor(String protocol, String host, int port) {
        this.protocol = protocol;
        this.host = host;
        this.port = port;

        serverAddress = protocol + host + ":" + port;
    }

    @Override
    public String getServerAddress() {
        return serverAddress;
    }

    @Override
    public void accept() {
        if (isAccepting()) {
            log.info("is already accept!");
            return;
        }
        int poolSize = Runtime.getRuntime().availableProcessors() + 1;
        executor = new ThreadPoolExecutor(poolSize, poolSize * poolSize, 600L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));
        try {
            InetAddress inetAddress = InetAddress.getByName(host);
            serverSocket = new ServerSocket(port, 256, inetAddress);
            running = true;
        } catch (IOException e) {
            log.warn(e.getMessage(), e);
            throw new RuntimeException(e);
        }
        // 注册服务地址
        //serviceRegistry.register(getServerAddress());

        executor.execute(new Runnable() {
            @Override
            public void run() {
                RpcProtocol rpcProtocol = RpcProtocol.ProtocolRegistry.getRpcProtocol(protocol);
                while (isAccepting()) {
                    try {
                        // block accept
                        Socket client = serverSocket.accept();
                        log.debug("client: {}", client);
                        executor.execute(new ServiceTask(rpcProtocol, rpcProcessor, client));
                    } catch (IOException e) {
                        log.warn(e.getMessage(), e);
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                        throw new RuntimeException(e);
                    }
                }
            }
        });
    }

    @Override
    public boolean isAccepting() {
        return running;
    }

    @Override
    public void close() throws IOException {
        log.info("close...");
        if (isAccepting()) {
            running = false;
            // 取消注册服务地址
            //serviceRegistry.unregister(getServerAddress());
            if (!serverSocket.isClosed()) {
                serverSocket.close();
            }
            if (!executor.isShutdown()) {
                executor.shutdown();
            }
        }
    }

    static class ServiceTask implements Runnable {
        RpcProtocol rpcProtocol;
        RpcProcessor rpcProcessor;
        Socket client;

        public ServiceTask(RpcProtocol rpcProtocol, RpcProcessor rpcProcessor, Socket client) {
            this.rpcProtocol = rpcProtocol;
            this.rpcProcessor = rpcProcessor;
            this.client = client;
        }

        @Override
        public void run() {
            InputStream input = null;
            OutputStream output = null;
            try {
                input = client.getInputStream();
                // 获取请求
                RpcRequest rpcRequest = rpcProtocol.decode(input, RpcRequest.class);
                // 处理请求
                RpcResponse rpcResponse = rpcProcessor.process(rpcRequest);
                // 返回处理结果
                byte[] bytes = rpcProtocol.encode(rpcResponse);
                output = client.getOutputStream();
                output.write(bytes);
                output.flush();
            } catch (IOException e) {
                log.error(e.getMessage(), e);
            } finally {
                IOUtils.close(output);
                IOUtils.close(input);
                IOUtils.close(client);
            }
        }
    }

}
